<?php
/*
 *  This file is part of Urd.
 *
 *  Urd is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *  Urd is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program. See the file "COPYING". If it does not
 *  exist, see <http://www.gnu.org/licenses/>.
 *
 * $LastChangedDate: 2009-01-04 14:49:53 +0100 (Sun, 04 Jan 2009) $
 * $Rev: 411 $
 * $Author: gavinspearhead $
 * $Id: queue.php 411 2009-01-04 13:49:53Z gavinspearhead $
 */


// This is an include-only file:
if (!defined('ORIGINAL_PAGE')) die('This file cannot be accessed directly.');


$pathq = realpath(dirname(__FILE__));
require_once "$pathq/../functions/autoincludes.php";


define ('MOVE_UP', 1);
define ('MOVE_DOWN', -1);


function swap(&$a, $x, $y)
{
    $tmp = $a[$x];
    $a[$x] = $a[$y];
    $a[$y] = $tmp;
}


function update_download_position(DatabaseConnection $db, action $item, $position)
{
    $command = $item->get_command_code();
    if ($command == COMMAND_DOWNLOAD || $command == COMMAND_DOWNLOAD_ACTION) {
        $id = $item->get_args();
        $sql = "UPDATE downloadinfo SET \"position\" = '$position' WHERE \"ID\"='$id'";
        $db->execute_query($sql);
    }
}


function swap_prio(DatabaseConnection $db, &$a, $x, $y, $username)
{
    $p1 = $a[$x]->get_priority();
    $p2 = $a[$y]->get_priority();
    $a[$y]->set_priority($p1, $username);
    $a[$x]->set_priority($p2, $username);
    update_queue_priority($db, $a[$x]->get_dbid(), $p2);
    update_queue_priority($db, $a[$y]->get_dbid(), $p1);
}


class queue {
    private $qq;
    private $max_size;

    function __construct ($max_size=0)
    {
        assert(is_numeric($max_size));
        $this->qq = array();
        $this->max_size = (int)$max_size;
    }
    function __destruct() 
    {
/* empty still */ 
    }
    function size() 
    {
        return count($this->qq); 
    }
    function is_empty() 
    {
        return count($this->qq) == 0; 
    }
    function get_queue() {
        $qq = array();
        foreach($this->qq as $q) {
            $row['id'] = $q->get_id();
            $row['username'] = $q->get_username();
            $row['priority'] = $q->get_priority();
            $row['command'] = $q->get_command();
            $row['args'] = $q->get_args();
            $row['queue_time'] = $q->get_queue_time();
            $row['pause'] = $q->is_paused();

            $qq[] = $row;
        }
        return $qq;
    }
    
        
    function update_all_download_position(DatabaseConnection $db)
    {
        $sql = 'UPDATE downloadinfo SET "position" = \'0\'';
        $db->execute_query($sql);
        $cnt = 0;

        foreach($this->qq as $q) {
            $cnt++;
            update_download_position($db, $q, $cnt);
        }
    }

    function move(DatabaseConnection $db, $cmd, $arg, $username, $direction)
    {
        $prev = $next = NULL;
        $cnt = 0;
        foreach($this->qq as $k=>$q) {
            if ($next !== NULL) {
                swap ($this->qq, $k, $next);
                swap_prio ($db, $this->qq, $k, $next, $username);
                $cnt++;
            } else if ($q->is_equal($cmd, $arg)) {
                if ($q->match_username($username)) {
                    if ($direction == MOVE_UP) {
                        if ($prev === NULL) 
                            continue; // no previous to swap with
                        swap($this->qq, $k, $prev);
                        swap_prio ($db, $this->qq, $k, $prev, $username);
                        $cnt++;
                    } else if ($direction == MOVE_DOWN) 
                        $next = $k;
                } else {
                    ; // do nothing
                }
            }
            $prev = $k;
        }
        if($cnt == 0)
            throw new exception ('Item not found',ERR_ITEM_NOT_FOUND);
        else
            $this->update_all_download_position($db);
    }   
        
    function push(action $item,DatabaseConnection $db, $increase_counter=TRUE, $priority=NULL) 
    { 
        assert(is_bool($increase_counter));
        if ($priority != NULL) {
            assert(is_numeric($priority));
            $item->set_priority($priority, $item->get_username());
            update_queue_priority($db, $item->get_dbid(), $priority);
        }
        return $this->add_prio ($item, $db, $increase_counter); 
    }
    function push_top(action $item, DatabaseConnection $db, $increase_counter=TRUE) 
    { 
        assert(is_bool($increase_counter));
        $item->set_priority(1, $item->get_username());
        return $this->add_prio ($item, $db, $increase_counter);
    }
    function move_top(DatabaseConnection$db, $index, $username)
    {
        assert(is_numeric($index));
        foreach($this->qq as $k=>$q) {
            if ($q->get_id() == $index) {
                if ($q->match_username($username)) {
                    unset($this->qq[$k]);
                    array_unshift($this->qq, $q);
                    $this->update_all_download_position($db);
                    return;
                } else
                    throw new exception('Not allowed', ERR_ACCESS_DENIED);
            }
        }
        throw new exception ('Item not found', ERR_ITEM_NOT_FOUND);
    }
    function has_equal(action $a)
    {
        foreach ($this->qq as $q) {
            if ($q->is_equal($a->get_command_code(), $a->get_args()))
                return TRUE;
        }
        return FALSE;

    }
    function get_ids_all($username)
    {
        $keys = array();
        foreach ($this->qq as $k => $q) {
            if ($q->match_username($username)) {
                $keys[] = $k;
            }
        }
        return $keys;

    }
    function get_ids_cmd($cmd, $arg, $username)
    {
        $keys = array();
        foreach ($this->qq as $k => $q) {
            if ($q->is_equal($cmd, $arg) && $q->match_username($username)) {
                $keys[] = $k;
            }
        }
        return $keys;
    }
    function get_ids_action($action_id, $username)
    {
        assert (is_numeric($action_id));
        $keys = array();
        foreach ($this->qq as $k => $q) {
            if ($action_id == $q->get_id()){
                $keys[] = $k;
            }
        }
        return $keys;
    }
    function delete_ids(DatabaseConnection $db, array $ids, $username, $delete_db=FALSE)
    {
        assert(is_bool($delete_db));
        $cnt = 0;
        foreach($ids as $id) {
            if (isset($this->qq[$id])) {
                if ($this->qq[$id]->match_username($username)) {
                    if ($delete_db) {
                        $status = QUEUE_CANCELLED;
                        update_queue_status($db, $this->qq[$id]->get_dbid(), $status);
                    }
                    unset($this->qq[$id]);
                    $cnt ++;
                } else {
                    throw new exception ('Not allowed', ERR_ACCESS_DENIED);
                }
            } else 
                throw new exception ('Queue item not found', ERR_ITEM_NOT_FOUND);
        }
        return $cnt;
    }
    function get_queue_item($id)
    {
        assert(is_numeric($id));
        if (isset($this->qq[$id]))
            return $this->qq[$id];
        else 
            throw new exception ('Queue item not found', ERR_ITEM_NOT_FOUND);
    
    }

/*  function delete_all(DatabaseConnection $db, $username, $delete_db=FALSE)
    {
        assert(is_bool($delete_db));
        $kk = array();
        foreach ($this->qq as $k => $q) {
            if ($q->match_username($username)) {
                $kk[] = $k;
                if ($delete_db === TRUE) {
                    $status = QUEUE_CANCELLED;
                    update_queue_status($db, $q->get_dbid(), $status);
                }
            }
        }
        foreach ($kk as $k) {
            unset ($this->qq[$k]);
        }
        $this->update_all_download_position($db);
    }*/
    function delete_cmd(DatabaseConnection $db, $cmd, $arg, $username, $delete_db = FALSE)
    {
        assert(is_bool($delete_db));
        $kk = array();
        foreach ($this->qq as $k => $q) {
            if ($q->is_equal($cmd, $arg)) {
                $kk[] = $k;
            }
        }
        if (empty($kk)) 
            return FALSE;

        foreach ($kk as $k) {
            if (!$this->qq[$k]->match_username($username)) 
                throw new exception ('Not allowed', ERR_ACCESS_DENIED);
            $item = $this->qq[$k];
            unset($this->qq[$k]);
            if($delete_db === TRUE) {
                $status = QUEUE_CANCELLED;
                update_queue_status($db, $item->get_dbid(), $status);
            }
        }
        $this->update_all_download_position($db);
        return TRUE;
    }
    function get_preview_action()
    {
        foreach ($this->qq as $q) {
            if ($q->get_preview() && !$q->is_paused()) {
                return $q;
            }
        }
        return FALSE;
    }
        
    function delete(DatabaseConnection $db, $action_id, $username, $delete_db = FALSE)
    {
        assert(is_bool($delete_db) && is_numeric($action_id));
        $kk = NULL;
        foreach ($this->qq as $k => $q) {
            if ($action_id == $q->get_id()){
                $kk = $k;
                break;
            }
        }
        if ($kk === NULL) 
            return FALSE;
        if (!$this->qq[$kk]->match_username($username)) 
            throw new exception ('Not allowed', ERR_ACCESS_DENIED);
        $item = $this->qq[$kk];
        if($delete_db === TRUE) {
            $status = QUEUE_CANCELLED;
            update_queue_status($db, $item->get_dbid(), $status);
        }
        unset($this->qq[$kk]);
        $this->update_all_download_position($db);
        return TRUE;
    }
    function pause(DatabaseConnection $db, $action_id, $do_pause, $username)// $do_pause == true then pause, else continue (unpause)
    {
        assert(is_bool($do_pause) && is_numeric($action_id));
        $status = $do_pause ? QUEUE_PAUSED : QUEUE_QUEUED;
        foreach ($this->qq as $q) {
            if ($action_id == $q->get_id()){
                $rv = $q->pause($db, $do_pause, $username);
                $cmd = $q->get_command();
                update_queue_status ($db, $q->get_dbid(), $status, NULL, NULL, NULL, $do_pause);
                if (compare_command($cmd, COMMAND_DOWNLOAD) ||compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
                    $dlid = $q->get_args();
                    update_dlinfo_status($db, $do_pause ? DOWNLOAD_PAUSED : DOWNLOAD_QUEUED, $dlid);
                }
                return TRUE;
            }
        }
        throw new exception ('item not found',ERR_ITEM_NOT_FOUND);
    }
    function pause_cmd(DatabaseConnection $db, $cmd, $arg, $do_pause, $username)// $do_pause == true then pause, else continue (unpause)
    {
        assert(is_bool($do_pause));
        $cnt = 0;
        $status = $do_pause ? QUEUE_PAUSED : QUEUE_QUEUED;
        foreach ($this->qq as $q) {
            if ($q->is_equal($cmd, $arg)) {
                try {
                    $rv = $q->pause($db, $do_pause, $username);
                    $cnt++;
                    update_queue_status ($db, $q->get_dbid(), $status, NULL, NULL, NULL, $do_pause);
                    if (compare_command($cmd, COMMAND_DOWNLOAD) ||compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
                        $dlid = $q->get_args();
                        update_dlinfo_status($db, $do_pause ? DOWNLOAD_PAUSED : DOWNLOAD_QUEUED, $dlid);
                    }

                } catch (exception $e) {
                    ;
                }
            }
        }
        if ($cnt == 0)
            throw new exception ('item not found',ERR_ITEM_NOT_FOUND);

    }
    function pause_all(DatabaseConnection $db, $do_pause, $username)// $do_pause == true then pause, else continue (unpause)
    {
        assert(is_bool($do_pause));
        $status = $do_pause ? QUEUE_PAUSED : QUEUE_QUEUED;
        foreach ($this->qq as $q) {
            try {
                $q->pause($db, $do_pause, $username);
                $cmd = $q->get_command();
                update_queue_status ($db, $q->get_dbid(), $status, NULL, NULL, NULL, $do_pause);
                if (compare_command($cmd, COMMAND_DOWNLOAD) ||compare_command($cmd, COMMAND_DOWNLOAD_ACTION)) { 
                    $dlid = $q->get_args();
                    update_dlinfo_status($db, $do_pause ? DOWNLOAD_PAUSED : DOWNLOAD_QUEUED, $dlid);
                }

            } catch (exception $e) {
                ;
            }
        }
    }
    protected function add_prio(action $item, DatabaseConnection $db, $increase_counter=TRUE)
    {
        assert(is_bool($increase_counter));
        if (($this->max_size > 0) && ($this->size() >= $this->max_size)) 
            throw new exception ('Queue full', ERR_QUEUE_FULL);
        if ($item->max_exceeded(MAX_REQUEUE_COUNT)) {
            $status = QUEUE_CANCELLED;
            update_queue_status($db, $item->get_dbid(), $status);
            write_log('Item queued too often, task cancelled: ' . "{$item->get_command()} {$item->get_args()}", LOG_NOTICE);
            return FALSE;
        }
        $item->inc_counter();
        if ($increase_counter === TRUE) { // false if we cancel a thread and push it back to the queue
            echo_debug('pushing on the queue', DEBUG_SERVER);
            $status = QUEUE_QUEUED;
            $dir = $item->get_dlpath();
            $dbid = insert_queue_status($db, $item->get_id(), "{$item->get_command()} {$item->get_args()}", $status, $item->get_command_code(), $item->get_username(), 0, '', $item->get_priority());
            if ($dbid !== FALSE) {
                $item->set_dbid($dbid);
                echo_debug("pushing on the queue $dbid", DEBUG_SERVER);
            } else 
                throw new exception ('Queueing failed', ERR_QUEUE_FAILED);
        } else 
            $dbid = $item->get_dbid();

        $prio = $item->get_priority();
        $temp_ql = $temp_qm = array();
        $status = $item->is_paused() ? QUEUE_PAUSED : QUEUE_QUEUED;
        update_queue_status ($db, $dbid, $status, NULL, NULL, NULL, $item->is_paused());
        
        foreach ($this->qq as $q) {
            if ($q->get_priority() <= $prio)
                $temp_ql[] = $q;
            else 
                $temp_qm[] = $q;
        }
        $temp_ql[] = $item;
        $this->qq = array_merge($temp_ql, $temp_qm);
        $this->update_all_download_position($db);
        return $dbid;
    }
/*  not used
    function pop(DatabaseConnection $db)
    {
        if($this->size() > 0) {
            $q = array_shift($this->qq);
            update_download_position($db, $q, 0);
            return $q;  
        } else 
            throw new exception ('Queue empty', ERR_QUEUE_EMPTY);
            
    }
 */
    function top($mayhave_nntp=TRUE, array $not_these=array(), $mayhave_db_intensive= TRUE)
    {
        assert(is_bool($mayhave_nntp));
        if ($this->is_empty())
            return FALSE;
        if ($mayhave_nntp === TRUE) {
            $first = NULL;
            foreach ($this->qq as $q) {
                if ($q->is_paused() === FALSE) {
                    if (in_array($q->get_id(), $not_these)) continue; // skip the ones we already tried
                    if ($mayhave_db_intensive === FALSE && $q->db_intensive()) continue; // skip, if it is a db intensive job and we may not start one
                    if ($q->need_nntp() === TRUE) // find the first with a nntp connection
                        return $q;
                    else if ($first === NULL)
                        $first = $q;
                }
            }
            return (($first !== NULL) ? $first:FALSE);
        } else {
            foreach ($this->qq as $q) {
                if (($q->need_nntp() === FALSE) && ($q->is_paused() === FALSE) && ($mayhave_db_intensive === TRUE || $q->db_intensive() === FALSE))
                    return $q;
            }
        }
        return FALSE;
    }
    function set_priority(DatabaseConnection $db, $action_id, $username, $priority)
    { 
        assert(is_numeric($action_id) && is_numeric($priority));
        $kk = NULL;
        foreach ($this->qq as $k => $q) {
            if ($action_id == $q->get_id()) {
                $q->set_priority($priority, $username);
                update_queue_priority($db, $q->get_dbid(), $priority);
                $kk = $k;
            }
        }
        if ($kk === NULL)
            throw new exception('item not found',ERR_ITEM_NOT_FOUND);
        $item = $this->qq[$kk];
        unset ($this->qq[$kk]);
        $prio = $item->get_priority();

        $temp_ql = $temp_qm = array();
        foreach ($this->qq as $q) {
            if ($q->get_priority() <= $prio)
                $temp_ql[] = $q;
            else 
                $temp_qm[] = $q;
        }
        $temp_ql[] = $item;
        $this->qq = array_merge($temp_ql, $temp_qm);
        $this->update_all_download_position($db);
        return TRUE;
    }
} //queue


?>
